package ddz.db.dao;

import com.kaka.notice.Proxy;
import com.kaka.notice.annotation.Model;
import com.kaka.util.Charsets;
import ddz.core.Table;
import ddz.db.Mybatis;
import ddz.db.mapper.TableSqlMapper;
import ddz.db.entity.TableInfo;
import ddz.db.redis.JedisFactory;
import ddz.db.redis.KryoSerializer;
import ddz.db.redis.Serializer;
import org.apache.ibatis.session.SqlSession;
import redis.clients.jedis.Jedis;

import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;


@Model
public class TableDao extends Proxy {

    private Serializer<Table> serializer = new KryoSerializer<>();
    private Queue<TableInfo> add_update_queue = new ArrayBlockingQueue<>(10000);

    private byte[] key(long uid) {
        return ("desk:k:" + uid).getBytes(Charsets.utf8);
    }

    private TableInfo getDeskInfoById(long id) {
        try (SqlSession session = Mybatis.getInstance().getSqlSessionFactory().openSession()) {
            TableSqlMapper tableSqlMapper = session.getMapper(TableSqlMapper.class);
            return tableSqlMapper.getDeskInfoById(id);
        }
    }

    private int insertDeskInfo(TableInfo info) {
        try (SqlSession session = Mybatis.getInstance().getSqlSessionFactory().openSession()) {
            TableSqlMapper tableSqlMapper = session.getMapper(TableSqlMapper.class);
            int result = tableSqlMapper.insertDeskInfo(info);
            session.commit();
            return result;
        }
    }

    private void batchInsertDeskInfo(List<TableInfo> list) {
        if (list == null || list.isEmpty()) return;
        try (SqlSession session = Mybatis.getInstance().getSqlSessionFactory().openSession()) {
            TableSqlMapper tableSqlMapper = session.getMapper(TableSqlMapper.class);
            tableSqlMapper.batchInsertDeskInfo(list);
            session.commit();
        }
    }

    private void updateDeskInfo(TableInfo info) {
        try (SqlSession session = Mybatis.getInstance().getSqlSessionFactory().openSession()) {
            TableSqlMapper tableSqlMapper = session.getMapper(TableSqlMapper.class);
            tableSqlMapper.updateDeskInfo(info);
            session.commit();
        }
    }

    public Table getDesk(long id) {
        Table table = null;
        byte[] keyBytes = key(id);
        byte[] valBytes = null;
        try (Jedis jedis = JedisFactory.getFactory().getJedis()) {
            valBytes = jedis.get(keyBytes);
        }
        if (valBytes != null && valBytes.length > 0) {
            table = serializer.deserialize(valBytes);
        }
        if (table == null) {
            TableInfo info = this.getDeskInfoById(id);
            if (info == null) return null;
            table = info.toDesk();
            try (Jedis jedis = JedisFactory.getFactory().getJedis()) {
                jedis.setex(keyBytes, 3600, serializer.serialize(table));
            }
        }
        return table;
    }

    public Table insertOrUpdateDesk(Table table) {
        byte[] keyBytes = key(table.getId());
        try (Jedis jedis = JedisFactory.getFactory().getJedis()) {
            jedis.setex(keyBytes, 3600, serializer.serialize(table));
        }
        TableInfo deskInfo = table.toDeskInfo();
        add_update_queue.add(deskInfo);
        return table;
    }

    private void processAddOrUpdateQueue(int count) {
        int size = add_update_queue.size();
        if (size < count) {
            count = size;
        }
        if (count == 0) return;
        Map<Long, Integer> idIdxMap = new HashMap<>(count);
        List<TableInfo> list = new ArrayList<>(count);
        for (int i = 0; i < count; i++) {
            TableInfo info = add_update_queue.poll();
            if (!idIdxMap.containsKey(info.getId())) {
                idIdxMap.put(info.getId(), list.size());
                list.add(info);
            } else {
                int idx = idIdxMap.get(info.getId());
                list.set(idx, info);
            }
        }
        this.batchInsertDeskInfo(list);
    }

    public void processAddOrUpdateQueue(boolean flush) {
        if (flush) {
            while (!this.add_update_queue.isEmpty()) {
                processAddOrUpdateQueue(50);
            }
        } else {
            processAddOrUpdateQueue(50);
        }
    }

}
